package com.surfilter.massdata.spark.util.ipcheck;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;
import java.util.UUID;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.json.JSONObject;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.collect.Maps;
import com.surfilter.massdata.spark.bean.ipcheck.BeianAllocate;
import com.surfilter.massdata.spark.bean.ipcheck.BeianSource;
import com.surfilter.massdata.spark.bean.ipcheck.IpBean;
import com.surfilter.massdata.spark.bean.ipcheck.UploadAllocate;
import com.surfilter.massdata.spark.bean.ipcheck.UploadSource;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.Global;
import com.surfilter.massdata.spark.util.JdbcUtil;
import com.surfilter.massdata.spark.util.JdbcUtilNew;

import scala.Tuple2;
import scala.collection.Iterator;

/**
 * @author hutao
 * @date 创建时间：2017年5月22日 上午9:05:29 @Description：
 * @version 1.0
 * 
 **/
public class IpCheckUtil {
	private static final Log log = LogFactory.getLog(IpCheckUtil.class);
	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy_MM_dd");
	static String date = sdf.format(new Date());

	public static String genBr7003PartitionName() {
		return "IP_CLASH_" + date;
	}

	public static String genBr7004PartitionName() {
		return "IP_CLASH_DETAIL_" + date;
	}

	public static String getUUID() {
		return UUID.randomUUID().toString().replace("-", "");
	}

	/**
	 * @Description:获取本省单位
	 * @return
	 */
	public static Map<Long, String> getBsdwMap() {
		Map<Long, String> bsdwMap = Maps.newHashMap();
		String shengId = Global.getConfig("surfilter.massdata.shengid");
		if (null != shengId) {
			String querySql = "SELECT ID FROM SY0022 WHERE SFYX = 1 AND TO_CHAR(SHENGID) = " + shengId;
			Connection conn = JdbcUtilNew.getConnection();
			PreparedStatement ps = null;
			try {
				ps = conn.prepareStatement(querySql);
				ResultSet rs = ps.executeQuery();
				while (rs.next()) {
					bsdwMap.put(rs.getLong(1), "");
				}
				if (rs != null) {
					rs.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			} finally {
				try {
					if (ps != null)
						ps.close();
					if (conn != null)
						conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}
		return bsdwMap;
	}

	/**
	 * @Description:获取运营商ID
	 * @return
	 */
	public static Map<String, String> getOptorIdsMap() {
		Map<String, String> optoridsMap = Maps.newHashMap();
		String[] ids = Global.getConfig("surfilter.ipcheck.stard.filter.bbdw").split(",");
		if (null != ids) {
			for (String id : ids) {
				optoridsMap.put(id, "");
			}
		}
		return optoridsMap;
	}

	/**
	 * @Description:获取接入企业经营许可证号
	 * @return
	 */
	public static Map<String, Long> getBr3010Map() {
		Map<String, Long> br3010Map = Maps.newHashMap();
		String query_br3010_sql = "select id, jyxkzbh from br3010";
		Connection conn = JdbcUtilNew.getConnection();
		PreparedStatement ps = null;
		try {
			ps = conn.prepareStatement(query_br3010_sql);
			ResultSet rs = ps.executeQuery();
			while (rs.next()) {
				Long id = rs.getLong(1);
				String bh = rs.getString(2);
				br3010Map.put(bh, id);
			}
			if (rs != null) {
				rs.close();
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				if (ps != null)
					ps.close();
				if (conn != null)
					conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return br3010Map;
	}
	
	public static  Map<Long, String>  getBr3010IdMap() {
		Map<Long, String> br3010Map = Maps.newHashMap();
		String query_br3010_sql = "select id from br3010";
		Connection conn = JdbcUtilNew.getConnection();
		PreparedStatement ps = null;
		try {
			ps = conn.prepareStatement(query_br3010_sql);
			ResultSet rs = ps.executeQuery();
			while (rs.next()) {
				br3010Map.put(rs.getLong(1), "");
			}
			if (rs != null) {
				rs.close();
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				if (ps != null)
					ps.close();
				if (conn != null)
					conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return br3010Map;
	}
	
	/**
	 * @Description:将DataFrame转换成javaRdd<IpBean>
	 * 
	 * @param resultFrame
	 * @return
	 */
	public static JavaRDD<IpBean> getIpBeanRdd(DataFrame resultFrame) {
		JavaRDD<IpBean> rdd = resultFrame.toJavaRDD().map(new Function<Row, IpBean>() {
			private static final long serialVersionUID = 1L;
			ObjectMapper om = new ObjectMapper();
			ObjectWriter ow = om.writer();
			@Override
			public IpBean call(Row row) throws Exception {
				IpBean bean = new IpBean();
				bean.setStorageType(row.getString(0));
				bean.setTflag(row.getString(1));
				bean.setId(((BigDecimal) row.get(2)).longValue());
				if (bean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
					// 备案分配库
					bean.setBbdw_id(null == row.get(3) ? null : ((BigDecimal) row.get(3)).longValue());
					bean.setSbsj(null == row.get(4) ? null : row.getTimestamp(4).getTime());
					bean.setQsip(null == row.get(5) ? null : ((BigDecimal) row.get(5)).longValue());
					bean.setZzip(null == row.get(6) ? null : ((BigDecimal) row.get(6)).longValue());
					bean.setSfgn(null == row.get(7) ? null : ((BigDecimal) row.get(7)).longValue());
					bean.setFpfs(null == row.get(8) ? null : ((BigDecimal) row.get(8)).longValue());
					bean.setFpdx(null == row.get(9) ? null : ((BigDecimal) row.get(9)).longValue());
					bean.setSyfs(null == row.get(10) ? null : ((BigDecimal) row.get(10)).longValue());
					bean.setFprq(null == row.get(11) ? null : row.getTimestamp(11).getTime());
					bean.setYmtzfwq(null == row.get(12) ? null : ((BigDecimal) row.get(12)).intValue());
					bean.setDwmc(null == row.get(13) ? null : row.getString(13));
					bean.setDwxz(null == row.get(14) ? null : ((BigDecimal) row.get(14)).longValue());
					bean.setDwxzjb(null == row.get(15) ? null : ((BigDecimal) row.get(15)).longValue());
					bean.setDwfl(null == row.get(16) ? null : ((BigDecimal) row.get(16)).longValue());
					bean.setDwhyfl(null == row.get(17) ? null : ((BigDecimal) row.get(17)).longValue());
					bean.setJyxkzh(null == row.get(18) ? null : row.getString(18));
					bean.setShengid(null == row.get(19) ? null : ((BigDecimal) row.get(19)).longValue());
					bean.setShiid(null == row.get(20) ? null : ((BigDecimal) row.get(20)).longValue());
					bean.setXianid(null == row.get(21) ? null : ((BigDecimal) row.get(21)).longValue());
					bean.setXxdz(null == row.getString(22) ? null : row.getString(22));
					bean.setLxrxm(null == row.getString(23) ? null : row.getString(23));
					bean.setLxrdh(null == row.getString(24) ? null : row.getString(24));
					bean.setLxrdzyj(null == row.getString(25) ? null : row.getString(25));
					bean.setWgip(null == row.get(26) ? null : ((BigDecimal) row.get(26)).longValue());
					bean.setWgdz(null == row.getString(27) ? null : row.getString(27));
					bean.setBz(null == row.getString(28) ? null : row.getString(28));
					if (bean.getStorageType().equals(IpCheckConstants.UPLOAD)) {
						// 上报分配库额外的字段
						bean.setIpsss(null == row.get(29) ? null : ((BigDecimal) row.get(29)).longValue());
						bean.setJtiplx(null == row.get(30) ? null : ((BigDecimal) row.get(30)).intValue());
						bean.setIpssjf(null == row.get(31) ? null : row.getString(31));
						bean.setSourceIdcId(null == row.get(32) ? null : row.getString(32));
						bean.setReportIdcId(null == row.get(33) ? null : row.getString(33));
						bean.setReportType(null == row.get(34) ? null : ((BigDecimal) row.get(34)).intValue());
						bean.setLastModifyTime(null == row.get(35) ? null : row.getTimestamp(35).getTime());
						bean.setIsWarn(null == row.get(36) ? null : ((BigDecimal) row.get(36)).intValue());
						bean.setHouseCode(null == row.get(37) ? null : row.getString(37));
						bean.setIplx(null == row.get(38) ? null : ((BigDecimal) row.get(38)).intValue());
						bean.setLylx(null == row.get(39) ? null : ((BigDecimal) row.get(39)).intValue());
						bean.setFpLyId(null == row.get(40) ? null : ((BigDecimal) row.get(40)).longValue());
						bean.setLymc(null == row.get(41) ? null : row.getString(41));

					} else if (bean.getStorageType().equals(IpCheckConstants.BEIAN)) {
						bean.setIplx(null == row.get(29) ? null : ((BigDecimal) row.get(29)).intValue());
						bean.setLylx(null == row.get(30) ? null : ((BigDecimal) row.get(30)).intValue());
						bean.setFpLyId(null == row.get(31) ? null : ((BigDecimal) row.get(31)).longValue());
						bean.setLymc(null == row.get(32) ? null : row.getString(32));
					}
				} else if (bean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)) {
					// 备案来源库
					bean.setBbdw_id(null == row.get(3) ? null : ((BigDecimal) row.get(3)).longValue());
					bean.setLyid(null == row.get(4) ? null : ((BigDecimal) row.get(4)).longValue());
					bean.setQsip(null == row.get(5) ? null : ((BigDecimal) row.get(5)).longValue());
					bean.setZzip(null == row.get(6) ? null : ((BigDecimal) row.get(6)).longValue());
					bean.setBz(null == row.get(7) ? null : row.getString(7));

					if (bean.getStorageType().equals(IpCheckConstants.UPLOAD)) {
						bean.setSourceIdcId(null == row.get(8) ? null : row.getString(8));
						bean.setReportIdcId(null == row.get(9) ? null : row.getString(9));
						bean.setReportType(null == row.get(10) ? null : ((BigDecimal) row.get(10)).intValue());
						bean.setLastModifyTime(null == row.get(11) ? null : row.getTimestamp(11).getTime());
						bean.setIsWarn(null == row.get(12) ? null : ((BigDecimal) row.get(12)).intValue());

					}

				}
				
				String conflictMsg = "";
				if (bean.getStorageType().equals(IpCheckConstants.BEIAN)
						&& bean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)) {
					BeianSource bs = ConvertUtil.convertIpBeanToBeianSource(bean);
					conflictMsg = ow.writeValueAsString(bs);
				} else if (bean.getStorageType().equals(IpCheckConstants.BEIAN)
						&& bean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
					BeianAllocate ba = ConvertUtil.convertIpBeanToBeianAllocate(bean);
					conflictMsg = ow.writeValueAsString(ba);
				} else if (bean.getStorageType().equals(IpCheckConstants.UPLOAD)
						&& bean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)) {
					UploadSource us = ConvertUtil.convertIpBeanToUploadSource(bean);
					conflictMsg = ow.writeValueAsString(us);
				} else if (bean.getStorageType().equals(IpCheckConstants.UPLOAD)
						&& bean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
					UploadAllocate ua = ConvertUtil.convertIpBeanToUploadAllocate(bean);
					conflictMsg = ow.writeValueAsString(ua);
				}
				bean.setConflictMsg(conflictMsg);
				return bean;
			}
		});
		return rdd;
	}

	/**
	 * 
	 * @Description:将rdd进行过滤，当flag是true返回正确的IpBean的rdd,为flase返回错误的IpBean的rdd
	 * @param ipBeanRdd
	 * @param flag
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterIpBeanRdd(JavaRDD<IpBean> ipBeanRdd, final boolean flag) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				boolean resultFlag = validateIpProperties(ipBean);
				if (flag) {
					return resultFlag;
				} else {
					return !resultFlag;
				}
			}
		});
		return rdd;
	}
	/**
	 * @Description:过滤获取上报库
	 * @param ipBeanRdd
	 * @param flag
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterUploadIpBeanRdd(JavaRDD<IpBean> ipBeanRdd) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				String storageType=ipBean.getStorageType();
				if(null!=storageType&&storageType.equals(IpCheckConstants.UPLOAD)){
					return true;
				}
				return false;
			}
		});
		return rdd;
	}
	/**
	 * @Description:过滤获取备案库
	 * @param ipBeanRdd
	 * @param flag
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterBeianIpBeanRdd(JavaRDD<IpBean> ipBeanRdd) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				String storageType=ipBean.getStorageType();
				if(null!=storageType&&storageType.equals(IpCheckConstants.BEIAN)){
					return true;
				}
				return false;
			}
		});
		return rdd;
	}
	/**
	 * @Description:过滤上报来源库
	 * @param ipBeanRdd
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterUploadSourceIpBeanRdd(JavaRDD<IpBean> ipBeanRdd) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				String tfFlag=ipBean.getTflag();
				String storageType=ipBean.getStorageType();
				if(null!=storageType&&storageType.equals(IpCheckConstants.UPLOAD)&&null!=tfFlag&&tfFlag.equals(IpCheckConstants.IP_SOURCE_TYPE)){
					return true;
				}
				return false;
			}
		});
		return rdd;
	}
	/**
	 * @Description:过滤上报分配库
	 * @param ipBeanRdd
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterUploadAllocateIpBeanRdd(JavaRDD<IpBean> ipBeanRdd) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				String tfFlag=ipBean.getTflag();
				String storageType=ipBean.getStorageType();
				if(null!=storageType&&storageType.equals(IpCheckConstants.UPLOAD)&&null!=tfFlag&&tfFlag.equals(IpCheckConstants.IP_ALLOCATE_TYPE)){
					return true;
				}
				return false;
			}
		});
		return rdd;
	}
	/**
	 * @Description:校验IP属性
	 * @param ipBean
	 * @return
	 */
	public static boolean validateIpProperties(IpBean ipBean) {
		boolean flag = true;
		try {
			long qsip = ipBean.getQsip();
			long zzip = ipBean.getZzip();
			IpUtils.longToIP(zzip);
			IpUtils.longToIP(qsip);
			// 然后监测IP段是否合法
			if (qsip > zzip) {
				return false;
			}
			// 接着监测IP段是否过大
			long ipSegMaxLen = Long.parseLong(Global.getConfig("surfilter.ipcheck.seg.len.max", "200000"));
			if (zzip - qsip > ipSegMaxLen) {
				return false;
			}
			// 如果前面都满足，继续校验，监测IP分配属性是否正确
			if (ipBean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
				if (ipBean.getFpdx() == null || ipBean.getFpdx().equals(new Long(0l))) {
					if (ipBean.getFpfs().equals(new Long(2l))) {
						return false;
					}
				} else {
					if (!ipBean.getFpfs().equals(new Long(2l))) {
						return false;
					}
				}
			}
		} catch (Exception e) {
			flag = false;
		}
		return flag;
	}

	/**
	 * @Description:JavaRDD<IpBean>转换为JavaPairRDD<Long, Iterable
	 *                                                  <IpBean>>为了对ip进行分组
	 * 
	 * @param ipBeanRdd
	 * @return
	 */
	public static JavaPairRDD<Long, Iterable<IpBean>> transJavaPariRdd(JavaRDD<IpBean> ipBeanRdd) {
		// 首先将rdd转换成<Long,Ipbean>方便groupBykey
		JavaPairRDD<Long, IpBean> parirRdd = ipBeanRdd.flatMapToPair(new PairFlatMapFunction<IpBean, Long, IpBean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<Tuple2<Long, IpBean>> call(IpBean ipBean) throws Exception {
				ArrayList<Tuple2<Long, IpBean>> results = new ArrayList<Tuple2<Long, IpBean>>();
				Long qsip = ipBean.getQsip();
				Long zzip = ipBean.getZzip();
				while (qsip <= zzip) {
					results.add(new Tuple2<Long, IpBean>(qsip, ipBean));
					qsip++;
				}
				return results;
			}
		});
		return parirRdd.groupByKey();// 同一个IP，多个bean
	}

	/**
	 * @Description:将IP校验不合理的冲突数据入库
	 * @param df
	 * @param conflictType
	 * @throws Exception
	 */
	public static void writeValidateConflictToTable(DataFrame df) throws Exception {

		final String br7003InsertSql = "INSERT INTO BR7003(ID, START_IP, END_IP, CONFLICT_TYPE, CONFLICT_UNITS, FIND_TIME)"
				+ " values(?, ?, ?, ?, ?, to_date(?,'yyyy-MM-dd'))";
		final String br7004InsertSql = "INSERT INTO BR7004(ID, CONFLICT_ID, CONFLICT_SOURCE, SSISPID, SOURCE_CONTENT, CREATE_TIME) VALUES(BR7004_SE.NEXTVAL, ?, ?, ?, ?, to_date(?,'yyyy-MM-dd'))";

		df.foreachPartition(new JavaForeachPartitionDFFunc() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<Row> it) {
				Connection conn = null;
				PreparedStatement ps7003 = null;
				PreparedStatement ps7004 = null;
				try {
					conn = JdbcUtilNew.getConnection();
					conn.setAutoCommit(false);
					int n = 0;
					int batchSize = 5000;
					ps7003 = conn.prepareStatement(br7003InsertSql);
					ps7004 = conn.prepareStatement(br7004InsertSql);
					while (it.hasNext()) {
						Row row = it.next();
						String storageType = row.getAs("storageType");
						long  bbwd_id = row.getAs("bbdw_id");
						String uuid = getUUID();
						ps7003.setString(1, uuid);
						ps7003.setLong(2, Long.parseLong(row.getAs("qsip").toString()));
						ps7003.setLong(3, Long.parseLong(row.getAs("zzip").toString()));
						if (storageType.equals(IpCheckConstants.BEIAN)) {
							ps7003.setInt(4, ConflictType.IP_SEG_ERROR.getTypeValue());
						} else {
							ps7003.setInt(4, ConflictType.IP_SEG_ERROR2.getTypeValue());
						}

						ps7003.setString(5, "," + bbwd_id + ",");
						ps7003.setString(6, DateUtil.getCurrentymd(new Date()));
//						ps7003.execute();
						ps7003.addBatch();

						ps7004.setString(1, uuid);
						ps7004.setString(2, storageType);
						ps7004.setLong(3, bbwd_id);
						ps7004.setString(4, row.getAs("conflictMsg").toString());
						ps7004.setString(5, DateUtil.getCurrentymd(new Date()));
//						ps7004.execute();
						ps7004.addBatch();

						// TODO 日志
						if (++n % batchSize == 0) {
							ps7003.executeBatch();
							conn.commit();
							ps7003.clearBatch();
							
							ps7004.executeBatch();
							conn.commit();
							ps7004.clearBatch();
							n = 0;
						}
					}
					ps7003.executeBatch();
					conn.commit();
					ps7003.clearBatch();
					ps7004.executeBatch();
					conn.commit();
					ps7004.clearBatch();
					log.info("完成入库冲突入库.....");
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					if (null != ps7003) {
						try {
							ps7003.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
					if (null != ps7004) {
						try {
							ps7004.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
					if (conn != null) {
						try {
							conn.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
				}

			}
		});

	}
	public static void writeConflictToTable(DataFrame df) throws Exception {

		final String br7003InsertSql = "INSERT INTO BR7003(ID, START_IP, END_IP, CONFLICT_TYPE, CONFLICT_UNITS, FIND_TIME,SHI_ID)"
				+ " values(?, ?, ?, ?, ?, to_date(?,'yyyy-MM-dd'), ?)";
		final String br7004InsertSql = "INSERT INTO BR7004(ID, CONFLICT_ID, CONFLICT_SOURCE, SSISPID, SOURCE_CONTENT, CREATE_TIME) VALUES(BR7004_SE.NEXTVAL, ?, ?, ?, ?, to_date(?,'yyyy-MM-dd'))";

		df.foreachPartition(new JavaForeachPartitionDFFunc() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<Row> it) {
				Connection conn = null;
				PreparedStatement ps7003 = null;
				PreparedStatement ps7004 = null;
				try {
					conn = JdbcUtilNew.getConnection();
					conn.setAutoCommit(false);
					int count = 0;
					int batchSize = 5000;
					ps7003 = conn.prepareStatement(br7003InsertSql);
					ps7004 = conn.prepareStatement(br7004InsertSql);
					while (it.hasNext()) {
						Row row = it.next();
						String conflictId =  getUUID();
						String jsonStr=row.getAs("jsonStr");
						String storageType = row.getAs("storageType");
						String bbwds = row.getAs("bbdws");
						String regionId=row.getAs("regionIds");
						ps7003.setString(1, conflictId);
						ps7003.setLong(2, Long.parseLong(row.getAs("qsip").toString()));
						ps7003.setLong(3, Long.parseLong(row.getAs("zzip").toString()));
						ps7003.setInt(4, Integer.parseInt(row.getAs("conflictType").toString()));
						ps7003.setString(5, "," + bbwds + ",");
						ps7003.setString(6, DateUtil.getCurrentymd(new Date()));
//						
						String[] ssispIds = bbwds.split(",");
						String[] regionIds = regionId.split(",");
						StringBuilder shiId = new StringBuilder(",");
						String[] jsonArray = jsonStr.split("\001");
						if(ssispIds.length == regionIds.length){
							for(int i =0 ; i < ssispIds.length ; i++){   //自身属性冲突
								ps7004.setString(1, conflictId);
								String json = jsonArray[i];
								JSONObject jsonObj = null;
								try{
									jsonObj = new JSONObject(json);
								}catch(Exception e){
									System.err.println("jsonStr>>>>>>>>>>>>>>>"+json+";row!!!!>>>"+row.toString());
								}
								
								if(null!=jsonObj){
									ps7004.setString(2, null==jsonObj.get("storageType")?"-1":jsonObj.get("storageType").toString());
									if(jsonObj.has("ipsss")){
										String value=String.valueOf(jsonObj.get("ipsss"));
										if(!isEmpty(value)){
											shiId.append(value).append(",");
										}
									}
								}
								ps7004.setLong(3, Long.parseLong(ssispIds[i]));
								ps7004.setString(4, json);
								ps7004.setString(5, DateUtil.getCurrentymd(new Date()));
								ps7004.addBatch();
							}
						}else if(ssispIds.length < regionIds.length){
							for(int i = 0 ; i<regionIds.length ; i++){   //对比冲突
								ps7004.setString(1, conflictId);
								String json = jsonArray[i];
								JSONObject jsonObj = new JSONObject(json);
								if(null!=jsonObj){
									ps7004.setString(2, null==jsonObj.get("storageType")?"-1":jsonObj.get("storageType").toString());
									if(jsonObj.has("ipsss")){
										String value=String.valueOf(jsonObj.get("ipsss"));
										if(!isEmpty(value)){
											shiId.append(value).append(",");
										}
									}
								}
								ps7004.setLong(3, Long.parseLong(ssispIds[0]));
								ps7004.setString(4, json);
								ps7004.setString(5, DateUtil.getCurrentymd(new Date()));
								ps7004.addBatch();
							}
						}else if(ssispIds.length > regionIds.length){
							ps7004.setString(1, conflictId);
							ps7004.setString(2, storageType);
							ps7004.setLong(3, Long.parseLong(ssispIds[0]));
							String json = jsonArray[0];
							ps7004.setString(4, json);
							ps7004.setString(5, DateUtil.getCurrentymd(new Date()));
							ps7004.addBatch();
							ps7004.setString(1, conflictId);
							ps7004.setString(2, storageType);
							ps7004.setLong(3, Long.parseLong(ssispIds[1]));
							ps7004.setNull(4, Types.VARCHAR);
							ps7004.setString(5, DateUtil.getCurrentymd(new Date()));
							ps7004.addBatch();
							JSONObject jsonObj = new JSONObject(json);
							if(null!=jsonObj){
								if(jsonObj.has("ipsss")){
									String value=String.valueOf(jsonObj.get("ipsss"));
									if(!isEmpty(value)){
										shiId.append(value).append(",");
									}
								}
							}
						}
						
						if(!shiId.toString().equals(",")){
							ps7003.setString(7, shiId.toString());
						}else{
							ps7003.setNull(7, Types.VARCHAR);
						}
						ps7003.addBatch();
						if(++count % batchSize==0){
							ps7003.executeBatch();
							conn.commit();
							ps7003.clearBatch();
							ps7004.executeBatch();
							conn.commit();
							ps7004.clearBatch();
							count = 0;
						}
						
						
					}//end while
					ps7003.executeBatch();
					conn.commit();
					ps7003.clearBatch();
					ps7004.executeBatch();
					conn.commit();
					ps7004.clearBatch();
//					count = 0;
					
				} catch (Exception e) {
					
					e.printStackTrace();
				} finally {
					if (null != ps7003) {
						try {
							ps7003.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
					if (null != ps7004) {
						try {
							ps7004.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
					if (conn != null) {
						try {
							conn.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
				}

			}
		});

	}
	/**
	 * @Description:初始化分区
	 * @param tableName
	 * @throws Exception
	 */
	public static void initPartition(String tableName) throws Exception {
		Connection conn = null;
		PreparedStatement psForInitDeal = null;
		ResultSet psForInitDealRs = null;
		try {
			// 先看看分区表是否存在，不存在就创建，存在就把该分区表里的数据清空
			conn = JdbcUtilNew.getConnection();
			conn.setAutoCommit(false);
			String isPartitionExistsSql = "SELECT COUNT(1) FROM USER_TAB_PARTITIONS WHERE TABLE_NAME = ? AND PARTITION_NAME = ?";
			psForInitDeal = conn.prepareStatement(isPartitionExistsSql);
			psForInitDeal.setString(1, tableName);
			if (tableName.equals("BR7003")) {
				psForInitDeal.setString(2, IpCheckUtil.genBr7003PartitionName());
			} else {
				psForInitDeal.setString(2, IpCheckUtil.genBr7004PartitionName());
			}
			psForInitDealRs = psForInitDeal.executeQuery();
			psForInitDealRs.next();
			int count = psForInitDealRs.getInt(1);
			if (psForInitDeal != null)
				psForInitDeal.close();
			if (psForInitDealRs != null)
				psForInitDealRs.close();
			if (count == 1) {
//				之前是删除当日按的，现在是只留当天的
//				String deletePartitionDataSql = "";
//				if (tableName.equals("BR7003")) {
//					deletePartitionDataSql = "DELETE FROM " + tableName + " WHERE FIND_TIME = to_date(?,'yyyy-MM-dd')";
//				} else {
//					deletePartitionDataSql = "DELETE FROM " + tableName + " WHERE CREATE_TIME = to_date(?,'yyyy-MM-dd')";
//				}
//				psForInitDeal = conn.prepareStatement(deletePartitionDataSql);
//				psForInitDeal.setString(1, DateUtil.getCurrentymd(new Date()));
//				psForInitDeal.execute();
//				 conn.commit();
			} else {
				Calendar cal = Calendar.getInstance();
				cal.add(Calendar.DAY_OF_YEAR, +1);
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
				String dateStr = sdf.format(cal.getTime());
				String declarePartitionSql = "";
				//TODO 去掉分区TABLESPACE
				if (tableName.equals("BR7003")) {
					declarePartitionSql = "ALTER TABLE " + tableName + " ADD PARTITION "
							+ IpCheckUtil.genBr7003PartitionName() + " " + "VALUES LESS THAN (to_date('" + dateStr
							+ "','yyyy-MM-dd')) ";
				} else {
					declarePartitionSql = "ALTER TABLE " + tableName + " ADD PARTITION "
							+ IpCheckUtil.genBr7004PartitionName() + " " + "VALUES LESS THAN (to_date('" + dateStr
							+ "','yyyy-MM-dd')) ";
				}
				psForInitDeal = conn.prepareStatement(declarePartitionSql);
				psForInitDeal.execute();
				 conn.commit();
			}
		} catch (SQLException e) {
			 try {
			 conn.rollback();
			 } catch (SQLException e1) {
			 e1.printStackTrace();
			 }
			throw new Exception(e);
		} finally {
			if (null != psForInitDeal)
				psForInitDeal.close();
			if (null != psForInitDealRs)
				psForInitDealRs.close();
			if (null != conn) {
				conn.close();
			}
		}
	}
	/**
	 * @Description:清空表数据
	 * @param tableName
	 */
	public static void truncateTable(String tableName){
		String deleteSql = "TRUNCATE TABLE "+tableName;
		Connection conn=null;
		PreparedStatement ps = null;
		try{
			//TODO 将JdbcUtilNew改为JdbcUtil
			conn = JdbcUtil.getConnection();
			ps = conn.prepareStatement(deleteSql);
			ps.executeUpdate();
		}catch(Exception e){
			log.error(e.getMessage(),e);
		}finally{
			if (null != ps) {
				try {
					ps.close();
				} catch (SQLException e) {
					log.error(e.getMessage(),e);
				}
			}
			if (conn != null) {
				try {
					conn.close();
				} catch (SQLException e) {
					log.error(e.getMessage(),e);
				}
			}
		}
		
	}
	public static boolean isEmpty(String str) {
		if (str == null || str.trim().isEmpty()||"null".equals(str))
			return true;
		return false;
	}
}
